﻿// EventPusher负责将消息推送给远端的订阅者.

using System;
using System.Collections.Concurrent;
using Google.Protobuf;
using System.Threading;
using System.Linq;
using GSE.protos;
using GSE.Rpc;

namespace GSE
{
    public delegate int pushDataSender(byte[] data);
    public struct EventPusherConfig
    {
        /// <summary>
        /// 默认的固定推送频率: 每秒推送的事件数。如: freq等于20时,则每隔50毫秒推送一次, 一次可推送多个消息/事件.
        /// </summary>
        public UInt16 DefaultEventPushFrequency;

        public pushDataSender SendData;
    }

    /// <summary>
    /// 和事件/消息推送相关的配置
    /// </summary>
    public struct EventConfig
    {
        public UInt16 EventPushFrequency;
    }

    /// <summary>
    /// 订阅的消息/事件推送注册对象.
    /// </summary>
    public class EventPushReg
    {
        protected BlockingCollection<IMessage> eventQueues0;
        protected BlockingCollection<IMessage> eventQueues1;
        protected UInt16 incomingQueueNo = 0;
        protected Mutex incomingQueueSwitchLock;
        public EventPushReg()
        {
            eventQueues0 = new BlockingCollection<IMessage>();
            eventQueues1 = new BlockingCollection<IMessage>();
            incomingQueueSwitchLock = new Mutex();
        }

        protected ILogger logger
        {
            get { return Logger.Default; }
        }

        public void QueueEvent(IMessage evt)
        {
            incomingQueueSwitchLock.WaitOne();
            if (incomingQueueNo == 0)
            {
                eventQueues0.Add(evt);
            }
            else
            {
                eventQueues1.Add(evt);
            }
            incomingQueueSwitchLock.ReleaseMutex();
        }

        /// <summary>
        /// 将队列里的消息批量取出,用于推送.
        /// </summary>
        /// <returns></returns>
        public IMessage[] Bump()
        {
            BlockingCollection<IMessage> bumpQueue;
            incomingQueueSwitchLock.WaitOne();
            if (incomingQueueNo == 0)
            {
                bumpQueue = eventQueues0;
                incomingQueueNo = 1;
            }
            else
            {
                bumpQueue = eventQueues1;
                incomingQueueNo = 0;
            }
            incomingQueueSwitchLock.ReleaseMutex();
            var n = bumpQueue.Count;
            var m = new IMessage[n];
            for (var i = n; n > 0; n--)
            {
                m[i - n] = bumpQueue.Take();
            }
            return m;
        }
    }

    public class EventPusher
    {
        protected static ConcurrentDictionary<string, ConcurrentDictionary<byte, EventPushReg>> eventPushRegs;
        protected static EventPusherConfig cfg;
        protected static CancellationTokenSource taskCancellation;
        /// <summary>
        /// 远端订阅信息表.
        /// </summary>
        protected static ConcurrentDictionary<ushort, EventSubscription<IMessage>> subs;

        protected static ConcurrentDictionary<string, EventConfig> registrations;

        protected static ConcurrentDictionary<byte, Timer> taskRunners;
        protected static int nextSubSeriesID = 0;

        public EventPusher()
        {
        }

        public EventPusher(EventPusherConfig evtCfg)
        {
            cfg = evtCfg;
            eventPushRegs = new ConcurrentDictionary<string, ConcurrentDictionary<byte, EventPushReg>>();
            subs = new ConcurrentDictionary<ushort, EventSubscription<IMessage>>();
            taskCancellation = new CancellationTokenSource();
            registrations = new ConcurrentDictionary<string, EventConfig>();
            taskRunners = new ConcurrentDictionary<byte, Timer>();
        }

        protected ILogger logger
        {
            get { return Logger.Default; }
        }

        protected byte genSubSeriesID()
        {
            var id = Interlocked.Increment(ref nextSubSeriesID);
            if (id > 254)
            {
                throw new EventSubscriptionSeriesIDOverflowException("series ID exceeded 254");
            }
            return (byte)id;
        }

        /// <summary>
        ///  设置事件/消息推送频率.
        /// </summary>
        /// <param name="eventName"></param>
        /// <param name="freq"></param>
        /// <returns>之前的推送频率.</returns>
        public UInt16 SetFrequency(string eventName, UInt16 freq)
        {
            EventConfig cfg;
            if (!registrations.TryGetValue(eventName, out cfg))
            {
                throw new UnRegisteredEventException(eventName);
            }
            var oldFreq = cfg.EventPushFrequency;
            cfg.EventPushFrequency = freq;
            var regs = getEventSubRegs(eventName);
            byte[] seriesIDs;
            if (regs == null || regs.Count < 1)
            {
                return oldFreq;
            }
            seriesIDs = regs.Keys.ToArray<byte>();

            Timer t;
            foreach (var id in seriesIDs)
            {
                var exists = taskRunners.TryGetValue(id, out t);
                if (exists)
                {
                    t.Change(0, freq);
                }
            }
            return oldFreq;
        }

        public void RegisterSubscribableEvent(string name, EventConfig evtCfg)
        {
            if (evtCfg.EventPushFrequency < 1)
            {
                evtCfg.EventPushFrequency = cfg.DefaultEventPushFrequency;
            }
            registrations[name] = evtCfg;
        }

        /// <summary>
        /// 发布一个消息/事件.
        /// </summary>
        /// <param name="name"></param>
        /// <param name="evt"></param>
        public void Pub(string name, IMessage evt)
        {
            try
            {
                var regs = getEventSubRegs(name);
                foreach (var reg in regs)
                {
                    logger.Debug($"preparing sending event[{name}]: {evt}");
                    reg.Value.QueueEvent(evt);
                }
            }
            catch (Exception ex)
            {
                logger.Warn($"failed to pub event: {ex.Message}");
            }
        }

        protected ConcurrentDictionary<byte, EventPushReg> getEventSubRegs(string name)
        {
            ConcurrentDictionary<byte, EventPushReg> regs;
            var created = eventPushRegs.TryGetValue(name, out regs);
            if (!created)
            {
                throw new Exception($"no event push registerations found for \"{name}\"");
            }
            return regs;
        }

        protected EventPushReg getOrCreateEventPushReg(string name, byte seriesID)
        {
            ConcurrentDictionary<byte, EventPushReg> regs;
            var created = eventPushRegs.TryGetValue(name, out regs);
            if (!created)
            {// 尚未添加过
                var regs1 = new ConcurrentDictionary<byte, EventPushReg>();
                var added = eventPushRegs.TryAdd(name, regs1);
                if (!added)
                {// 已经被其它线程添加.
                    eventPushRegs.TryGetValue(name, out regs);
                }
                else
                {
                    regs = regs1;
                }
            }
            EventPushReg reg;
            created = regs.TryGetValue(seriesID, out reg);
            if (!created)
            {// 尚未添加过
                var reg1 = new EventPushReg();
                var added = regs.TryAdd(seriesID, reg1);
                if (!added)
                {// 已经被其它线程添加.
                    regs.TryGetValue(seriesID, out reg);
                }
                else
                {
                    reg = reg1;
                }
            }
            return reg;
        }

        /// <summary>
        /// 收到远端订阅消息的请求时,调用此方法创建对应的push任务.
        /// </summary>
        /// <param name="iSub">SubEvent对象</param>
        /// <returns>SubEventResponse</returns>
        public SubEventResponse OnNewSub(SubEvent sub)
        {
            EventConfig evtCfg;
            if (!registrations.TryGetValue(sub.Name, out evtCfg))
            {
                throw new UnRegisteredEventException(sub.Name);
            }
            var rsp = new SubEventResponse();
            var seriesID = genSubSeriesID();
            rsp.SeriesID = seriesID;
            var r = getOrCreateEventPushReg(sub.Name, seriesID);
            var t = new PushTask(r, sub.Name, (byte)rsp.SeriesID, cfg.SendData);
            var runner = new Timer(t.Run);
            runner.Change(0, 1000 / evtCfg.EventPushFrequency);
            var added = taskRunners.TryAdd(seriesID, runner);
            if (!added)
            {
                runner.Change(Timeout.Infinite, 1);
                rsp.Msg = "subscription series ID conflicts.";
                rsp.Status = EventRpcStatusTypes.Failed;
            }
            else
            {
                rsp.Status = EventRpcStatusTypes.Ok;
            }
            return rsp;
        }

        public CancelSubEventResponse OnUnSub(CancelSubEvent unSub)
        {
            var ok = false;
            var seriesID = (byte)unSub.SeriesID;
            EventPushReg r;
            foreach (var regs in eventPushRegs)
            {
                foreach (var reg in regs.Value)
                {
                    if (reg.Key == seriesID)
                    {
                        regs.Value.TryRemove(seriesID, out r);
                        Timer t;
                        var exists = taskRunners.TryRemove(seriesID, out t);
                        if (exists)
                        {
                            t.Change(Timeout.Infinite, 1);
                            t.Dispose();
                        }
                        ok = true;
                    }
                }
            }
            var rsp = new CancelSubEventResponse();
            if (!ok)
            {
                rsp.Status = EventRpcStatusTypes.Failed;
                rsp.Msg = "subscription does not exist.";
            }
            else
            {
                rsp.Status = EventRpcStatusTypes.Ok;
            }

            return rsp;
        }

        public void Stop()
        {
            foreach (var t in taskRunners.Values)
            {
                t.Change(Timeout.Infinite, 100000);
                t.Dispose();
            }
            taskCancellation.Cancel();
        }

    }
}
